package process;

import bean.SensorReading;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.junit.Test;

//TODO 实现10s温度没有下降则报警
public class Flink_Process_State_Ontimer {
    //TODO 使用process实现10s温度没有下降则报警
    public static class MyKeyedProcessFunc extends KeyedProcessFunction<Tuple,SensorReading,String>{
        //定义时间间隔
        private Integer interval;

        public MyKeyedProcessFunc(Integer interval) {
            this.interval = interval;
        }

        //TODO 声明温度状态用于存储每次的温度值
        private ValueState<Double> temp;

        //TODO 声明时间状态用于存放定时器时间
        private ValueState<Long> tsState;

        //状态初始化
        @Override
        public void open(Configuration parameters) throws Exception {
            temp = getRuntimeContext().getState(new ValueStateDescriptor<Double>("temp_state",Double.class,Double.MIN_VALUE));
            tsState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("ts_state",Long.class));
        }

        @Override
        public void processElement(SensorReading value, Context ctx, Collector<String> out) throws Exception {
            //获取上次温度值
            Double lastTemp = temp.value();

            //获取定时器时间状态
            Long lastTs = tsState.value();

            //获取当前传入温度
            Double currentTemp = value.getTemp();

            //定时时间,当前时间+interval
            long timerTs = ctx.timerService().currentProcessingTime() + interval * 1000L;


            //温度上升并且时间状态为Null，则注册定时器
            if (currentTemp >lastTemp && lastTs ==null){
                //注册定时器
                ctx.timerService().registerProcessingTimeTimer(timerTs);

                //更新温度和时间状态
                temp.update(currentTemp);
                tsState.update(ctx.timerService().currentProcessingTime());

            }
            //当温度下降时，删除定时器
            if (currentTemp <lastTemp && lastTs !=null){
                //删除定时器
                ctx.timerService().deleteProcessingTimeTimer(lastTs);

                //清空时间状态
                tsState.clear();

            }
        }

        //TODO 定时器触发
        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            //输出数据
            out.collect(ctx.getCurrentKey()+"连续10s温度没有下降");

            //清空时间状态
            tsState.clear();
        }
    }

    @Test
    public void test() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        env.socketTextStream("node193",9999)
                .flatMap(new FlatMapFunction<String, SensorReading>() {
                    @Override
                    public void flatMap(String value, Collector<SensorReading> out) throws Exception {
                        String[] split = value.split(",");
                        out.collect(new SensorReading(split[0],Long.parseLong(split[1]),Double.parseDouble(split[2])));
                    }
                })
                .keyBy(SensorReading::getId)
                .process(new KeyedProcessFunction<String, SensorReading, String>() {
                    private ValueState<Double> lastTemp;
                    private ValueState<Long> times;

                    @Override
                    public void open(Configuration configuration) throws Exception{
                        lastTemp=getRuntimeContext().getState(new ValueStateDescriptor<Double>("temp",Double.class));
                        times=getRuntimeContext().getState(new ValueStateDescriptor<Long>("time",Long.class));
                    }

                    @Override
                    public void processElement(SensorReading value, Context ctx, Collector<String> out) throws Exception {
                        Double last = lastTemp.value();
                        Long times1 = times.value();
                        Double nowTemp = value.getTemp();

                        long timerTs = ctx.timerService().currentProcessingTime() + 10000L;

                        if(last==null){
                            lastTemp.update(nowTemp);
                            ctx.timerService().deleteProcessingTimeTimer(timerTs);
                            times.clear();
                        }else if(nowTemp>=last && times1 == null){
                            ctx.timerService().registerProcessingTimeTimer(timerTs);
                            lastTemp.update(nowTemp);
                            times.update(ctx.timerService().currentProcessingTime());
                        }else if(nowTemp<last){
                            ctx.timerService().deleteProcessingTimeTimer(timerTs);
                            times.clear();
                        }

                        out.collect("nowTemp="+nowTemp);
                    }

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
                        out.collect(ctx.getCurrentKey()+"持续10s温度上升");
                        times.clear();
                    }
                })
                .print();
        env.execute();
    }
}
